package com.tdy.cdc.app.ods;

import com.alibaba.fastjson.JSONObject;
import com.tdy.cdc.app.BaseApp;
import com.tdy.cdc.util.DateFormatUtil;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.MyMysqlCdcDeserialization;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import lombok.extern.log4j.Log4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.nio.charset.Charset;

/**
 * @author NanHuang
 * @Date 2023/3/6
 */
@Log4j
public class OdsApp extends BaseApp {
    public static void main(String[] args) {
        new OdsApp().init(
                3001,
                "OdsApp"
        );
    }
    @Override
    public void invoke(StreamExecutionEnvironment env) {
        // 1 使用Flink-CDC读取MySQL的业务数据
        DataStreamSource<String> ds = readMySQL(env);
        // 2 写入Kafka
        writeToKafka(ds);
    }

    private void writeToKafka(DataStreamSource<String> ds) {
        ds.map(JSONObject::parseObject)
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject obj, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        String type = obj.getString("type");
                        JSONObject result = new JSONObject();
                        result.put("op",type);
                        result.put("table_name",obj.getString("table"));
                        if ("c".equals(type) || "u".equals(type) || "r".equals(type)){
                            result.putAll(obj.getJSONObject("after"));
//                            result.put("ds_date", DateFormatUtil.toYmdHms(System.currentTimeMillis()));
                            out.collect(result);
                        } else {
                            // TODO 数据删除操作
                        }
                    }
                })
                .sinkTo(FlinkSinkUtil.getKafkaSink(false,"ods_app"));
    }

    private DataStreamSource<String> readMySQL(StreamExecutionEnvironment env) {
        System.out.println(Charset.defaultCharset());
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.31.206")
                .port(3306)
                .databaseList("dwd_test")
                .tableList("dwd_test.dwd_mhs.*")
//                .tableList("sj_yuan.patient_.*,sj_yuan.tdy_list_.*")
//                .tableList("sj_yuan.hdsd00_.*")
                .username("root")
                .password("password")
//                .startupOptions(StartupOptions.latest())
                .deserializer(new MyMysqlCdcDeserialization())
                .build();
        return env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"mysql_cdc");//Flink-CDC读取数据
    }
}
